package com.bdqn.mr.mj;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapJoinMapper extends Mapper<LongWritable, Text, TableBean, NullWritable> {

    // <"01","小米"> <"02","华为"> <"03","格力">
    private Map<String, String> pdMap = null;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 在这里，给pdMap赋值
        pdMap = new HashMap<>();

        // 读缓存中的文件：采用 流 的方式去读取文件
        URI[] cacheFiles = context.getCacheFiles();
        URI cacheFile = cacheFiles[0];

        // 需要分布式文件系统的客户端，去读取上面的文件
        FileSystem fs = FileSystem.get(context.getConfiguration());
        // 读取文件的输入流
        FSDataInputStream fsDataInputStream = fs.open(new Path(cacheFile));

        // 把字节流包装为字符串的缓冲流，可以每次读取一行字符串
        BufferedReader reader = new BufferedReader(new InputStreamReader(fsDataInputStream, "utf-8"));

        String line = null;
        while (StringUtils.isNotEmpty(line = reader.readLine())) {
            // 01	小米
            String[] fields = line.split("\t");
            pdMap.put(fields[0], fields[1]);
        }

        // 关闭资源
        reader.close();
        fsDataInputStream.close();
        fs.close();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1001	01	1
        String line = value.toString();
        String[] fields = line.split("\t");

        bean.setOrderId(fields[0]);
        bean.setPid(fields[1]);
        bean.setAmount(Integer.parseInt(fields[2]));

        // 关键是这一步的map怎么构建
        bean.setpName(pdMap.get(bean.getPid()));

        context.write(bean, NullWritable.get());
    }

    public static final TableBean bean = new TableBean();
}
